package ex.datastream.sources;

import ex.vo.CityPojo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.csv.CsvReaderFormat;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.CsvMapper;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.CsvSchema;
import org.apache.flink.util.function.SerializableFunction;

import java.io.File;

public class FileDataSource {
    public FileSource<CityPojo> getFileSource(){
        SerializableFunction<CsvMapper, CsvSchema> schemaGenerator =
                mapper -> mapper.schemaFor(CityPojo.class)
                        .withColumnSeparator('|');

        CsvReaderFormat<CityPojo> csvFormat =
                CsvReaderFormat.forSchema(() -> new CsvMapper(), schemaGenerator, TypeInformation.of(CityPojo.class))
                        .withIgnoreParseErrors();//忽略错行

        FileSource<CityPojo> source =
                FileSource.forRecordStreamFormat(csvFormat, Path.fromLocalFile(new File("data/csv01"))).build();
        return source;
    }
}
